package Sink;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import utils.MySqlUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;

/**
 * Create by zhh on 2022/9/6.
 */
//Tuple2<String,Double>  String：域名   Double：流量
public class MySqlSink extends RichSinkFunction<Tuple2<String,Double>> {

    Connection connection;
    PreparedStatement insertPsmt;
    PreparedStatement updatePsmt;

    @Override
    public void open(Configuration parameters) throws Exception {

        connection = MySqlUtils.getConnection();
        insertPsmt = connection.prepareStatement("insert into zh_traffic(domain,traffic) values (?,?)");
        updatePsmt = connection.prepareStatement("update zh_traffic set traffic=? where domain =?");
    }


    @Override
    public void close() throws Exception {

        if (insertPsmt!=null) insertPsmt.close();
        if (updatePsmt!=null) updatePsmt.close();
        if (connection!=null) connection.close();
    }

    @Override
    public void invoke(Tuple2<String, Double> value, Context context) throws Exception {
        System.out.println("=======invoke======"+value.f0+"-->"+value.f1);
        updatePsmt.setDouble(1,value.f1);
        updatePsmt.setString(2,value.f0);
        updatePsmt.execute();
        if (updatePsmt.getUpdateCount()==0){//如果没有更新就插入
            insertPsmt.setString(1, value.f0);
            insertPsmt.setDouble(2,value.f1);
            insertPsmt.execute();
        }
    }
}
